/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.core;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.EnumSet;
import java.util.List;
import java.util.Properties;
import java.util.Scanner;
import java.util.UUID;

import lac.cnet.sddl.objects.ApplicationObject;
import lac.cnet.sddl.objects.Message;
import lac.cnet.sddl.udi.core.UniversalDDSLayerFactory;
import lac.cnet.sddl.udi.core.listener.UDIDataReaderListener;

import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import br.pucrio.inf.lac.konoha.hyuga.command.Command;
import br.pucrio.inf.lac.konoha.hyuga.command.CommandFactory;
import br.pucrio.inf.lac.konoha.hyuga.common.Grid;
import br.pucrio.inf.lac.konoha.hyuga.core.Scenario.ScenarioBuilder;

/**
 * The main entry point of the Hyuga platform.
 * 
 * This class boots the CEP Engine with the algorithm rules and configure the SDDL communication layer to receive and 
 * send data to applications. The bootstrap process reads the algorithm's parameter from a "configuration.properties".
 * This file includes key=value parameters to configure the monitored domain (scenario) and the data stream platform. 
 * 
 * Finally, Hyuga provides a command shell to interact and manage the platform.
 * 
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class Bootstrap {

	/** Algorithm parameters. */
	private Properties config;
	
	/** Algorithm Scenario. */
	private Scenario scenario;
	
	/** Data Stream. */
	private DDSLink dataStream;
	
	/** Hyuga (CEP) Service. */
	private HyugaService hyuga;
	
	/** Shell. */
	private Shell shell;
	
	/** Monitoring Agent. */
	private MonitoringAgent monAgent = MonitoringAgent.getInstance();
	
	/** LOGGER. */
    private static Logger LOGGER = Logger.getLogger(Bootstrap.class.getCanonicalName());

    /**
     * The Hyuga Platform constructor. It receives as input a properties files. This file contains domain values and
     * data stream properties, while the second property file contains the EPLs to be deployed in key=value format.
     * 
     * @param configProperties Property containing the algorithm and data stream parameters.
     */
    public Bootstrap(Properties configProperties) throws IllegalArgumentException {
        this.config = configProperties;
 		if (!valid(config)) throw new IllegalArgumentException();
	}
	
	/**
	 * Check if a Hyuga configuration file (in properties format) is valid.
	 * 
	 * @param properties The user parameters.
	 * @return true if its valid, false otherwise.
	 */
	private boolean valid(Properties properties) {
		boolean valid = true;

		if (properties == null) {
			valid = false;
			LOGGER.error("Properties file is null");
		}
		else if (properties == null || !properties.containsKey("eps") 	 || !properties.containsKey("minPts")
									|| !properties.containsKey("minlat") || !properties.containsKey("maxlat")
									|| !properties.containsKey("minlng") || !properties.containsKey("maxlng")
									|| !properties.containsKey("win") 	 || !properties.containsKey("dds")
									|| !properties.containsKey("deploy")) {
		    LOGGER.error("Missing a required parameter (eps, minPts, minlat, maxlat, minlng, maxlng, win, dds, or deploy)");
			valid = false;
		} else {
			String ddsProduct = (String) properties.get("dds");
			if (!ddsProduct.equalsIgnoreCase("OpenSplice") && !ddsProduct.equalsIgnoreCase("OSPL") && 
				!ddsProduct.equalsIgnoreCase("CoreDX")     && !ddsProduct.equalsIgnoreCase("RTI")) {
				LOGGER.error("Invalid DDS product: " + ddsProduct);
				valid = false;
			}
			
			if (properties.containsKey("distributed")) {
			    boolean distributed = Boolean.parseBoolean(properties.getProperty("distributed"));
			    
			    if (distributed) {
	                if (properties.getProperty("publish")   == null) valid = false;
	                if (properties.getProperty("subscribe") == null) valid = false;
	            }
			}

			if (properties.containsKey("deploy")) {
			    for (String epa : properties.getProperty("deploy").split(",")) {
			        if (!(epa.equalsIgnoreCase("ALL") || epa.equalsIgnoreCase("CELL") || 
                          epa.equalsIgnoreCase("CSV") || epa.equalsIgnoreCase("GRID"))) {
			            valid = false;
			            break;
			        }
			    }
			}
		}
		return valid;
	}
	
    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
	// Building stages
    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////

	/**
	 * Initiate the monitored domain configuration (scenario) through the properties values.
	 * 
	 * @throws IllegalArgumentException if a parameter is invalid. For example, maxlat < minlat, or ε < 0.
	 */
	public void buildScenario() throws IllegalArgumentException {
	    // Algorithm Inputs
        double minlat = Double.parseDouble(config.getProperty("minlat"));
        double maxlat = Double.parseDouble(config.getProperty("maxlat"));
        double minlng = Double.parseDouble(config.getProperty("minlng"));
        double maxlng = Double.parseDouble(config.getProperty("maxlng"));
        double eps    = Double.parseDouble(config.getProperty("eps"));
        double win    = Double.parseDouble(config.getProperty("win"));
        int    minPts = Integer.parseInt(config.getProperty("minPts"));
        
        // ε / √2 Partitions?
        if (config.containsKey("epsbysqrt") && Boolean.parseBoolean(config.getProperty("epsbysqrt"))) {
            eps = eps / Math.sqrt(2);
        }
        
        // Required Configuration        
        ScenarioBuilder scenarioBuilder = new ScenarioBuilder().latitude(minlat, maxlat)
                                                               .longitude(minlng, maxlng)
                                                               .eps(eps)
                                                               .minPts(minPts)
                                                               .timeWindow(win);

        // Are we using Cell Sub Division Strategy?
        if (config.containsKey("cellsub") && Boolean.parseBoolean(config.getProperty("cellsub")) &&
            config.containsKey("cellsubfunction")) {
            scenarioBuilder.usingCellSubDiv(true)
                           .decayFunction(config.getProperty("cellsubfunction"))
                           .cellSubDivision(Integer.parseInt(config.getProperty("cellsubdivision")))
                           .cellMinMinPts(Integer.parseInt(config.getProperty("cellminminpts")));
        } else {
            scenarioBuilder.usingCellSubDiv(false);
        }
        
        // Build the Scenario
        this.scenario = scenarioBuilder.build();
        
        // Hook scenario function to Grid
        if (scenario.hasDecayFunction()) {
            Grid.setDecayFunction(scenario.getDecayFunction());
        }
	}
	
	/**
	 * Initiate the data stream using the publish/subscribe topics.
	 */
    public void buildDataStream() {
        // Algorithm Input
        dataStream = DDSLink.getInstance(UniversalDDSLayerFactory.convertStrToSupportedDDSVendor(config.getProperty("dds")));
        
        // Instance name
        dataStream.setDeployName(config.getProperty("name", "DEFAULT" + UUID.randomUUID().toString()));

        // It is a distributed deploy? If yes, we deploy custom publish/subscribe topics, otherwise we use the standard topics.
        boolean distributed = Boolean.parseBoolean(config.getProperty("distributed"));
        if (distributed) {
            for (String pubTopic : config.getProperty("publish").split(","))   { 
                dataStream.createSpatialTopic(pubTopic); 
            }
            for (String subTopic : config.getProperty("subscribe").split(",")) { 
                dataStream.createSpatialTopic(subTopic); 
            }
        }
        
        // Default Topics
        dataStream.createDefaultTopics();
    }

	/** 
	 * Build the Hyuga (algorithm) EPN on top of the stream-processing engine.
	 */
	public void deployHyuga() {
        // Internal variables
        boolean debug = Boolean.parseBoolean(config.getProperty("debug"));
        
        // Instatiate Hyuga (Mapper, CEP engine, etc)
        hyuga = new HyugaService(dataStream, scenario, debug);       

        // Which EPAs are we going to deploy?
        String epasToDeploy = config.getProperty("deploy");
        EnumSet<EPA> epaSet = EPA.getSet(epasToDeploy);
     
        hyuga.deploy(epaSet);
        
        // Link the Data Stream with Hyuga
        for (String subTopic : config.getProperty("subscribe").split(",")) {
            // Are we using filtered topics?
            boolean filter = config.containsKey("filter") ? Boolean.parseBoolean(config.getProperty("filter").trim()) : false;
            if (filter) {
                String filterLogic = config.getProperty("filterlogic");
                List<String> param = new ArrayList<String>();
                
                for (String p : config.getProperty("filterparam").split(",")) { param.add(p); }
                
                dataStream.addFilteredListener(subTopic, hyuga, filterLogic, param);
            } else {
                dataStream.addListener(subTopic, hyuga); 
            }
        }
        
        // Local Deploy?
        if (epasToDeploy.equalsIgnoreCase("ALL")) {
            hyuga.setLocalDeploy(true);
        } else {
            hyuga.setLocalDeploy(false);      
        }
	}
	
    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
	// Shell
    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////

	/** Prepare Shell */
    public void prepareShell() {
        boolean debug = Boolean.parseBoolean(config.getProperty("debug"));
        
        shell = new Shell(new CommandFactory(config, hyuga));
        shell.printShellInfo(scenario, debug);
        
        dataStream.addListener("Meta", shell);
        shell.loop();
    }
    
    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
	// CLI and Main Routines
    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
	
	/**
     * Main Routine.
     * 
     * @param args is a String array indicating the algorithm parameters files (configuration to use and epls to load).
     *             If null, the algorithm tries to load the default files (configuration.properties and epl.properties) 
     *             in the current directory.
     * @throws IOException
     */
	public static void main(String[] args) {
	    try {
            PropertyConfigurator.configure(readPropertyFile("log4j.properties"));
        } catch (IOException e) {
            System.err.print("Error at reading log4j file");
        }
	    
        // Loading application parameters
        Properties config = null;
        try {
            if (args.length == 0) {
                LOGGER.info("Loading default parameters (configuration.properties)");
                config = readPropertyFile("configuration.properties");
            } else { 
                LOGGER.info("Loading parameters from " + args[0]);
                config = readPropertyFile(args[0]);
            }
        } catch (IOException ex) {
            LOGGER.error("Error at reading the configuration file");
            System.exit(-1);
        }

        // Booting the system!
        Bootstrap boot = new Bootstrap(config);
        try {
            boot.buildScenario();
            boot.buildDataStream();
            boot.deployHyuga();
            boot.prepareShell();
        } catch (IllegalArgumentException ex) {
            LOGGER.error(ex.getMessage());
	    }
	}
	
    private static Properties readPropertyFile(String filename) throws IOException {
	    Properties properties = new Properties();
	    
        // Try to load indicated property file
        File propertyFile = new File(filename);
        
        if (!propertyFile.exists()) {
            LOGGER.warn("File " + filename + " not found in the current directory");
            LOGGER.warn("Trying to load " + filename + " using the classpath");
            
            InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(filename);
            if (is != null) {
                properties.load(is);
            } else {
                LOGGER.error("File " + filename + " does not exist");
                throw new FileNotFoundException();
            }
        } else {
            FileInputStream fi = new FileInputStream(propertyFile);
            LOGGER.info("Loaded parameters from file: " + propertyFile.getName());
            properties.load(fi);
            fi.close();
        }
        
        return properties;
	}

    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Gets and Sets
    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////

    public Scenario getScenario() {
        return scenario;
    }
    
    public Properties getProperties() {
        return config;
    }

    public HyugaService getHyuga() {
        return hyuga;
    }
    
    public DDSLink getDataStream() {
        return dataStream;
    }
    
    public void setDataStream(DDSLink dataStream) {
        this.dataStream = dataStream;
    }
}
